package com.fs.controller;

import com.fs.producer.Producer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.SendResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;

/**
 * @author ：hyq
 * @date ：Created in 2020/5/20 14:21
 * @description：
 * @modified By：
 */
@RestController
public class ProducerTest {
    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private Producer producer;
    @GetMapping("send")
    public void getTest() throws InterruptedException, ExecutionException {
        for (int i = 0; i < 10; i++) {
            int id = 1;
            SendResult result = producer.syncSendOrderly(id);
            logger.info("[testSyncSend][发送编号：[{}] 发送队列：[{}]]", i, result.getRecordMetadata().partition());
        }

        // 阻塞等待，保证消费
//        new CountDownLatch(1).await();
    }

    @GetMapping("sendAlarm")
    public void getTestSend() throws InterruptedException, ExecutionException {
        for (int i = 0; i < 10; i++) {
            producer.send(i);

        }
    }

    /**
     * 测试kafka 结合 spring 注解事务发送
     * 发生异常后数据不会发送成功
     */
    @GetMapping("traction")
    public void tractionTest(){
        try {
            producer.testTransactionalAnnotation();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @GetMapping("test")
    public String test(){
        return "33333";
    }
}
